﻿using System;
using System.Collections.Generic;
using System.Net;
using System.Net.Sockets;
using System.Reflection;
using System.Text;
using System.Threading;
using Microsoft.Office.Project.EPMAuditing.AuditListenerCatalog;
using Microsoft.Office.Project.EPMAuditing.Library;
using System.IO;
using System.Globalization;

namespace Microsoft.Office.Project.EPMAuditing.ReceiverService
{
    class AuditingWorkerThreads
    {

        private EPMAuditingConfigFile _configFile;
        private EPMAuditingConfig _config;

        private Tracing _tracing = null;
        private bool _udpThreadClosing;

        private ManualResetEvent _pause;

        private GetRefreshListenerListCallback _getRefreshListenerList;
        private SetRefreshListenerListCallback _setRefreshListenerList;

        private GetStartNewFileCallback _getStartNewFileCallback;
        private SetStartNewFileCallback _setStartNewFileCallback;

        //private Queue<string> EPMAuditMessages;
        private Queue<MemoryStream> EPMAuditMessages;
        private Object EPMAuditMessageQueueLock = new Object();

        private Dictionary<Guid, MessageSegmentCollection> _buildingMessages;

        private Queue<MessageSegment> _messagePackets;

        /// <summary>
        /// Constructor
        /// </summary>
        /// <param name="tracing">Pointer to the active tracing component.</param>
        /// <param name="pause">Allows for the service to tell the thread to wait for instructions.</param>
        /// <param name="getRefreshListenerList">Callback location for allowing the service to tell the thread to sync now.</param>
        /// <param name="setRefreshListenerList">Callback location for allowing the service to tell the thread to sync now.</param>
        public AuditingWorkerThreads(Tracing tracing, ManualResetEvent pause,
            GetRefreshListenerListCallback getRefreshListenerList, SetRefreshListenerListCallback setRefreshListenerList,
            GetStartNewFileCallback getStartNewFileCallback, SetStartNewFileCallback setStartNewFileCallback)
        {
            _tracing = tracing;
            _pause = pause;
            _getRefreshListenerList = getRefreshListenerList;
            _setRefreshListenerList = setRefreshListenerList;

            _getStartNewFileCallback = getStartNewFileCallback;
            _setStartNewFileCallback = setStartNewFileCallback;


            // If the configuration data doesn't initialize, then terminate the thread
            if (!InitializeData())
            {
                throw new ArgumentNullException("_config", "Unable to read the configuration information.");
            }

            //EPMAuditMessages = new Queue<string>(10);
            EPMAuditMessages = new Queue<MemoryStream>(10);
            _buildingMessages = new Dictionary<Guid, MessageSegmentCollection>();
            _messagePackets = new Queue<MessageSegment>(25);
        }

        #region Initialization Methods

        private bool InitializeData()
        {
            // Read configuration file
            string fullPath = Assembly.GetExecutingAssembly().Location;
            _configFile = new EPMAuditingConfigFile(_tracing, string.Format(CultureInfo.InvariantCulture, "{0}{1}", 
                fullPath.Substring(0, fullPath.LastIndexOf(@"\", StringComparison.OrdinalIgnoreCase) + 1),
                EPMAuditingConfigFile.DEFAULT_CONFIG_FILENAME));
            _config = _configFile.AuditingConfiguration;
            if (_config == null)
                return false;

            return true;
        }


        #endregion

        #region Worker Threads

        /// <summary>
        /// Entry point for staring up the thread for process changes.  Calls the initialization methods, if they fail
        /// the worker thread will exit.
        /// </summary>
        public void UDPListenerWorkerMethod()
        {
            _tracing.WriteTrace(_tracing.MainSwitch.TraceInfo, "Starting the UDPListenerWorker thread");

            //DateTime wakeupTime = DateTime.Now.AddSeconds(_config.WakeupInterval);
            _tracing.WriteTrace(_tracing.MainSwitch.TraceInfo, "Entering heartbeat cycle for UDPListenerWorker thread");

            UdpClient client = null;

            try
            {
                IPEndPoint endPoint = new IPEndPoint(IPAddress.Parse(_config.NetAuditingServerIP), _config.NetAuditingServerPort);
                //IPEndPoint endPoint = new IPEndPoint(IPAddress.Parse(EPMAuditingConsts.NetAuditingIP), EPMAuditingConsts.NetAuditingPort);
                client = new UdpClient(endPoint);
                client.BeginReceive(new AsyncCallback(ReceiveCallback), client);
                _tracing.WriteTrace(_tracing.MainSwitch.TraceInfo, "1");

                do
                {
                    Thread.Sleep(50);
                    // Block if the service is paused or is shutting down.
                    _pause.WaitOne();

                } while (true);
            }
            catch (ThreadAbortException)
            {
                if (client != null)
                {
                    _udpThreadClosing = true;
                    client.Close();
                    Thread.Sleep(200);
                }
                Thread.ResetAbort();
                _tracing.WriteTrace(_tracing.MainSwitch.TraceInfo, "UDPListenerWorker thread abort signaled");
            }
            catch (Exception ex)
            {
                _tracing.WriteTrace(_tracing.MainSwitch.TraceError, string.Format(CultureInfo.InvariantCulture, "UDPListenerWorker thread, exception thrown: {0}", ex.Message));
            }

            _tracing.WriteTrace(_tracing.MainSwitch.TraceInfo, "Exiting the UDPListenerWorker thread");
        }

        public void ReceiveCallback(IAsyncResult ar)
        {
            if (ar != null)
            {
                UdpClient client = ar.AsyncState as UdpClient;

                // Need to check if the worker thread closed, otherwise could be in a race condition and throw an exception
                if (!_udpThreadClosing)
                {
                    IPEndPoint endPoint = new IPEndPoint(IPAddress.Parse(_config.NetAuditingServerIP), _config.NetAuditingServerPort);

                    Byte[] receiveBytes = client.EndReceive(ar, ref endPoint);
                    MemoryStream ms = new MemoryStream(receiveBytes);

                    MessageSegment msgSeg = new MessageSegment(ms);
                    lock (EPMAuditMessageQueueLock)
                    {
                        _messagePackets.Enqueue(msgSeg);
                    }
                    _tracing.WriteTrace(_tracing.MainSwitch.TraceInfo, string.Format(CultureInfo.InvariantCulture, "READ Message UID:{0}, {1} of {2}",
                        msgSeg.MessageUid.ToString(), msgSeg.SegmentNumber.ToString(CultureInfo.InvariantCulture), 
                        msgSeg.SegmentCount.ToString(CultureInfo.InvariantCulture)));

                    client.BeginReceive(new AsyncCallback(ReceiveCallback), client);

                }
            }
        }


        public void JoinPackets()
        {
            int queueMesssages = 0;
            MessageSegment msgSeg = null;

            do
            {
                lock (EPMAuditMessageQueueLock)
                {
                    queueMesssages = _messagePackets.Count;
                    if (queueMesssages > 0)
                    {
                        msgSeg = _messagePackets.Dequeue();
                    }
                }

                if (queueMesssages > 0)
                {
                    if (!_buildingMessages.ContainsKey(msgSeg.MessageUid))
                    {
                        // MessageSegments doesn't exist, create it
                        MessageSegmentCollection msgSegList = new MessageSegmentCollection(msgSeg.MessageUid, msgSeg.SegmentCount);
                        _buildingMessages.Add(msgSeg.MessageUid, msgSegList);
                        _tracing.WriteTrace(_tracing.MainSwitch.TraceInfo, "Create MessageSegments");
                    }

                    // add the new piece
                    _buildingMessages[msgSeg.MessageUid].Add(msgSeg);
                }
            } while (queueMesssages > 0);
        }

        public void MoveCompleteMessagesToQueue()
        {
            List<Guid> messagesToRemove = new List<Guid>();

            foreach (KeyValuePair<Guid, MessageSegmentCollection> kvp in _buildingMessages)
            {
                if (_buildingMessages[kvp.Key].MessageComplete)
                    messagesToRemove.Add(kvp.Key);
            }

            foreach (Guid removeMessageUid in messagesToRemove)
            {
                MemoryStream myMemStr = _buildingMessages[removeMessageUid].Message;

                _tracing.WriteTrace(_tracing.MainSwitch.TraceInfo, string.Format(CultureInfo.InvariantCulture, "Message Complete, size: {0}",
                    myMemStr.Length.ToString(CultureInfo.InvariantCulture)));
                // Push the object on to the queue
                EPMAuditMessages.Enqueue(myMemStr);
                _buildingMessages[removeMessageUid].Clear();
                _buildingMessages.Remove(removeMessageUid);

            }

        }

        public void AuditWriterWorkerMethod()
        {
            _tracing.WriteTrace(_tracing.MainSwitch.TraceInfo, "Starting the AuditWriterWorkerMethod thread");
            bool refreshListenerList = false;
            bool getStartNewFileCallback = false;

            AuditListenerCollection als = new AuditListenerCollection(_config.Listeners, _tracing);

            DateTime wakeupTime = DateTime.Now.AddSeconds(_config.WakeupIntervalSeconds);
            _tracing.WriteTrace(_tracing.MainSwitch.TraceInfo, "Entering heartbeat cycle for AuditWriterWorkerMethod");


            try
            {
                do
                {
                    refreshListenerList = _getRefreshListenerList();
                    if (refreshListenerList)
                    {
                        _tracing.WriteTrace(_tracing.MainSwitch.TraceInfo, "AuditWriterWorkerMethod thread received RefreshListenerList Signal");
                        _setRefreshListenerList(false);
                        refreshListenerList = false;
                        // Refresh from config file
                        if (!InitializeData())
                        {
                            throw new ArgumentNullException("_config", "Unable to re-read the configuration information.");
                        }
                        // Refresh the listeners from the new config data
                        als.UpdateAuditListeners(_config.Listeners, _tracing);
                    }

                    getStartNewFileCallback = _getStartNewFileCallback();
                    if (getStartNewFileCallback)
                    {
                        _tracing.WriteTrace(_tracing.MainSwitch.TraceInfo, "AuditWriterWorkerMethod thread received StartNewFile Signal");
                        _setStartNewFileCallback(false);
                        getStartNewFileCallback = false;

                        foreach (AuditListener al in als)
                        {
                            if (Utilities.DerivedFromBaseFileAuditListener(al.GetType()))
                            {
                                BaseFileAuditListener bfa = (BaseFileAuditListener)al;
                                bfa.StartNewFile();
                            }
                        }
                    }


                    if (DateTime.Now > wakeupTime)
                    {
                        int queueMesssages = 0;

                        do
                        {
                            JoinPackets();
                            MoveCompleteMessagesToQueue();

                            MemoryStream ms = null;
                            if (EPMAuditMessages.Count > 0)
                            {
                                ms = EPMAuditMessages.Dequeue();
                                //receiveString = EPMAuditMessages.Dequeue();
                                queueMesssages = EPMAuditMessages.Count;
                            }
                            else
                                queueMesssages = 0;

                            if (ms != null)
                                if (ms.Length > 0)
                                {
                                    // writes to each audit listener that is available.
                                    MemoryStream tempMemoryStream = null;
                                    foreach (AuditListener al in als)
                                    {
                                        ms.Position = 0;
                                        tempMemoryStream = new MemoryStream();
                                        ms.WriteTo(tempMemoryStream);
                                        tempMemoryStream.Position = 0;
                                        _tracing.WriteTrace(_tracing.MainSwitch.TraceInfo, string.Format(CultureInfo.InvariantCulture, "al.write al type={0}, MemStr.Len={1}", 
                                            al.GetType().ToString(), tempMemoryStream.Length.ToString(CultureInfo.InvariantCulture)));
                                        al.Write(tempMemoryStream);
                                        if (tempMemoryStream != null)
                                            tempMemoryStream.Dispose();
                                        //al.Write(receiveString);
                                    }
                                    // TODO: Dispose of the MemoryStream
                                }
                        } while (queueMesssages > 1);

                        wakeupTime = DateTime.Now.AddSeconds(_config.WakeupIntervalSeconds);
                    }

                    // Wait before cycling.
                    Thread.Sleep(1000);
                    // Block if the service is paused or is shutting down.
                    _pause.WaitOne();
                } while (true);
            }
            catch (ThreadAbortException)
            {
                Thread.ResetAbort();
                _tracing.WriteTrace(_tracing.MainSwitch.TraceInfo, "AuditWriterWorkerMethod thread abort signaled");
            }

            als.Dispose();
            _tracing.WriteTrace(_tracing.MainSwitch.TraceInfo, "Exiting the AuditWriterWorkerMethod thread");
        }


        #endregion


    }

}
